from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import NoSuchTableError, OperationalError, ProgrammingError
from typing import List, Tuple, Any, Iterator, Dict
import os
import sqlparse
import re
import json
import csv
from openpyxl import Workbook
import sys
import datetime
import inspect
import os


class SqlxDatabaseError(Exception):
    """
    #### 自定义异常类, 用于抛出数据库错误异常
    """

    pass


class ValueTypeError(Exception):
    """
    #### 自定义异常类, 用于抛出值类型错误异常
    """

    pass


class ValueNoneError(Exception):
    """
    #### 自定义异常类, 用于抛出值不能为空异常
    """

    pass


def check_value_type(value, expected_type, keyword):
    """
    #### 验证传入的值是否为指定类型, 如果不是则抛出ValueTypeError异常。
    - value: 需要验证的值。
    - expected_type: 期望的值类型。
    - keyword: 用于描述该值的关键字, 用于在异常信息中提供更具体的上下文。
    - raises ValueTypeError: 如果值的类型与期望类型不匹配, 则抛出此异常, 并附带指定的错误信息。
    """
    if not isinstance(value, expected_type):
        raise ValueTypeError(
            f"{keyword} 类型错误, 期望类型为 {expected_type}, 实际类型为 {type(value)}"
        )


def value_not_none(
    value: Any, keyword: str = "", is_diy: bool = False, diy_msg: str = None
):
    """
    #### 验证传入的值是否为None, 如果为None则抛出ValueNoneError异常。
    - value: 需要验证的值。
    - keyword: 用于描述该值的关键字, 用于在异常信息提供不存在的关键字。默认为空字符串。
    - is_diy: 是否为自定义异常, 默认值为False, 如果为True, 则抛出自定义异常
    - raises ValueNoneError: 如果值为None, 则抛出此异常, 并附带错误信息。
    """
    if not value:
        # 如果是自定义异常, 则抛出自定义异常
        if is_diy:
            raise ValueNoneError(f"{keyword} {diy_msg}")
        # 否则抛出默认异常
        raise ValueNoneError(f"{keyword} 不能为空")


# 定义一个Result类, 用于封装查询结果
class Result:
    """
    #### Result类, 用于封装查询结果
    - status: 表示操作是否成功, 类型为布尔值
    - result: 表示操作结果, 类型为 List[Tuple[Any]], 默认为空列表
    - rowcount: 表示操作影响的行数, 类型为整数, 默认为0, -1 表示查询操作没有影响的行数
    - message: 用于需要记录信息的场景, 类型为字符串, 默认为空字符串
    #### 方法
    - is_success(): 判断操作是否成功, true表示成功, false表示失败
    - results_iter(): 返回一个迭代器, 用于遍历查询结果中的每一行数据。
    - to_dicts_iter(): 将结果转换为字典生成器
    - to_jsons(): 将查询结果序列化为 JSON 格式
    - export_to_csv(): 将查询结果导出为 CSV 文件
    - export_to_excel(): 将查询结果导出为 Excel 文件
    """

    # 初始化方法, 用于创建Result对象
    def __init__(
        self,
        status: bool = False,
        result: List[Tuple[Any]] = [],
        rowcount: int = 0,
        message: str = "",
    ):
        # status属性, 表示操作是否成功, 类型为布尔值
        self.status = status
        # result属性, 表示操作结果, 类型为列表包元组, 默认为空列表
        self.result = result if result is not None else []
        # rowcount属性, 表示操作影响的行数, 类型为整数, 默认为0, -1 表示查询操作没有影响的行数
        self.rowcount = rowcount
        # message属性, 用于需要记录信息的场景, 类型为字符串, 默认为空字符串
        self.message = message

    # 特殊方法__repr__, 用于返回对象的字符串表示, 便于调试和日志记录
    def __repr__(self):
        """
        #### 返回对象的字符串表示
        - return: str 类型
        """
        return f"Result(status={self.status}, result={self.result}, rowcount={self.rowcount}, message={self.message})"

    # is_success 方法, 用于判断操作是否成功, true表示成功, false表示失败
    def is_success(self):
        """
        #### 判断操作是否成功
        - return: bool 类型 True 表示成功, False 表示失败
        """
        return self.status

    def results_iter(self) -> Iterator[Tuple[Any]]:
        """
        #### 返回一个迭代器, 用于遍历查询结果中的每一行数据。
        - return 一个迭代器, 每次迭代返回一个包含查询结果中一行数据的元组。
        """
        for item in self.result:
            yield item

    def to_dicts_iter(self, keys: List[str]) -> Iterator[Dict[str, Any]]:
        """
        #### 将结果转换为字典生成器
        - keys: 字典的键列表, 用于指定元组中每个字段的名称
        - return: Iterator[Dict[str, Any]]
        """
        # 检查每个元组的长度是否与键列表的长度一致
        expected_length = len(keys)
        for item in self.results_iter():
            if len(item) != expected_length:
                raise ValueError(
                    f"键列表的长度 ({len(keys)}) 与元组的长度 ({len(item)}) 不匹配。"
                    f"请检查键列表和数据元组是否一致。"
                    f"\n键列表: {keys}"
                    f"\n数据元组: {item}"
                )
            yield dict(zip(keys, item))

    def to_jsons(self, keys: List[str]) -> str:
        """
        #### 将查询结果序列化为 JSON 格式
        - keys: 字典的键列表, 用于指定元组中每个字段的名称
        - return: JSON 格式的字符串
        """
        # 先将结果转换为字典列表
        dict_list = list(self.to_dicts_iter(keys))
        # 将字典列表序列化为 JSON 格式的字符串
        return json.dumps(dict_list, ensure_ascii=False, indent=4, default=str)

    def export_to_csv(self, output_file: str = None, keys: List[str] = None) -> bool:
        """
        #### 将查询结果导出为 CSV 文件
        - output_file: 输出文件路径
        - keys: 字典的键列表, 用于指定元组中每个字段的名称
        - return: bool 类型, True 表示导出成功, False 表示导出失败
        """
        try:
            # 检查输出文件路径和键列表是否为空
            if not output_file or not keys:
                raise ValueError("输出文件路径和键列表不能为空。")

            # 检查每个元组的长度是否与键列表的长度一致
            expected_length = len(keys)
            for item in self.results_iter():
                if len(item) != expected_length:
                    raise ValueError(
                        f"键列表的长度 ({len(keys)}) 与元组的长度 ({len(item)}) 不匹配。"
                        f"请检查键列表和数据元组是否一致。"
                        f"\n键列表: {keys}"
                        f"\n数据元组: {item}"
                    )

            # 将结果写入 CSV 文件
            with open(output_file, "w", newline="", encoding="utf-8") as csvfile:
                writer = csv.DictWriter(csvfile, fieldnames=keys)
                writer.writeheader()
                for row in self.to_dicts_iter(keys):
                    writer.writerow(row)

            return True

        except Exception as e:
            print(f"导出到 CSV 文件失败: {e}")
            return False

    def export_to_excel(self, output_file: str, keys: List[str]) -> bool:
        """
        #### 将查询结果导出为 Excel 文件
        - output_file: 输出文件路径
        - keys: 字典的键列表, 用于指定元组中每个字段的名称
        - return: bool 类型, True 表示导出成功, False 表示导出失败
        """
        try:
            # 检查输出文件路径和键列表是否为空
            if not output_file or not keys:
                raise ValueError("输出文件路径和键列表不能为空。")

            # 检查每个元组的长度是否与键列表的长度一致
            expected_length = len(keys)
            for item in self.results_iter():
                if len(item) != expected_length:
                    raise ValueError(
                        f"键列表的长度 ({len(keys)}) 与元组的长度 ({len(item)}) 不匹配。"
                        f"请检查键列表和数据元组是否一致。"
                        f"\n键列表: {keys}"
                        f"\n数据元组: {item}"
                    )

            # 将结果写入 Excel 文件
            workbook = Workbook()
            sheet = workbook.active
            # 写入表头
            sheet.append(keys)

            # 写入数据
            for row in self.to_dicts_iter(keys):
                sheet.append([row[key] for key in keys])
            workbook.save(output_file)

            return True

        except Exception as e:
            print(f"导出到 Excel 文件失败: {e}")
            return False


class Sqlx:
    """
    #### Sqlx类, 用于封装数据库操作, 自动管理事务, 并提供一些常用的数据库操作方法。
    - db_url: 数据库连接字符串
    - pool_size: 连接池中维护的最大连接数
    - max_overflow: 连接池中允许超过 pool_size 的临时连接数
    - pool_timeout: 获取数据库连接时等待的最长时间（秒）
    - pool_recycle: 连接的最大生命周期（秒）
    - enable_log: 是否启用日志, 默认值为 False
    #### 方法
    - newSqlx(db_url: str, pool_size: int = 5, max_overflow: int = 10, pool_timeout: int = 30, pool_recycle: int = 3600, enable_log: bool = False) -> Sqlx: 创建一个新的Sqlx对象
    - exec_sql(sql: str, params: List[Any] = None) -> Result: 执行 SQL 语句
    - exec_script(self, script_path: str, batch_size: int = 100) -> Result: 执行脚本文件
    - get_connection_info(self) -> dict: 获取数据库连接信息
    - truncate_table(self, table_name: str) -> bool: 清空表数据
    - close(self) -> None: 关闭数据库连接
    """

    def __init__(
        self,
        db_url: str,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_timeout: int = 30,
        pool_recycle: int = 3600,
        enable_log: bool = False,
    ):
        """
        #### 初始化数据库连接和连接池
        - db_url: 数据库连接字符串
        - pool_size: 连接池中维护的最大连接数 默认值为 5
        - max_overflow: 连接池中允许超过 pool_size 的临时连接数 默认值为 10
        - pool_timeout: 获取数据库连接时等待的最长时间（秒）默认值为 30 秒
        - pool_recycle: 连接的最大生命周期（秒）, 超过这个时间的连接将被关闭并重新创建, 设置为 -1 表示不关闭连接, 默认值为 3600 秒
        - enable_log: 是否启用日志, 默认值为 False
        - return: Sqlx 对象
        #### 示例:
        1. sqlite3 数据库连接字符串: sqlite:///./data/app.db (相对路径)
        2. sqlite3 数据库连接字符串: sqlite:////data/app.db (绝对路径)
        3. mysql 数据库连接字符串: mysql://user:password@host:port/dbname
        4. postgresql 数据库连接字符串: postgresql://user:password@host:port/dbname
        """
        try:
            # 检查数据库连接字符串是否为空
            value_not_none(db_url, "数据库连接字符串")
            # 检查连接池大小是否为空
            value_not_none(pool_size, "连接池大小")
            # 检查最大溢出连接数是否为空
            value_not_none(max_overflow, "最大溢出连接数")
            # 检查连接超时时间是否为空
            value_not_none(pool_timeout, "连接超时时间")
            # 检查连接最大生命周期是否为空
            value_not_none(pool_recycle, "连接最大生命周期")
            # 检查连接池大小是否为正整数
            if pool_size <= 0:
                raise ValueError("连接池大小必须为正整数, 切不能小于等于0")
            # 检查最大溢出连接数是否为正整数
            if max_overflow <= 0:
                raise ValueError("最大溢出连接数必须为正整数, 切不能小于等于0")
            # 检查连接超时时间是否为正整数
            if pool_timeout <= 0:
                raise ValueError("连接超时时间必须为正整数, 切不能小于等于0")
            # 创建数据库引擎
            self.engine = create_engine(
                db_url,
                pool_size=pool_size,
                max_overflow=max_overflow,
                pool_timeout=pool_timeout,
                pool_recycle=pool_recycle,
            )
            # 创建会话工厂
            self.Session = sessionmaker(bind=self.engine)
            # 创建会话
            self.session = self.Session()
            # 设置是否启用日志 self.enable_log 可以开启底层的日志
            self._enable_logger = enable_log
            # 实例化日志器
            self._logger = Logger()

        except Exception as e:
            # 如果连接失败,将engine设置为None, engine是数据库引擎
            self.engine = None
            # 如果连接失败, 将Session设置为None, Session是会话工厂
            self.Session = None
            # 如果连接失败, 将session设置为None, session是会话
            self.session = None
            # 打印
            if self._enable_logger:
                self._logger.error(f"实例化Sqlx失败: {e}")
            # 实例化失败, 抛出异常
            raise (f"实例化Sqlx失败: {e}")

    # exec_sql 方法, 用于执行 SQL 语句
    def exec_sql(
        self, sql: str, params: dict = None, is_return_data: bool = False
    ) -> Result:
        """
        #### 执行原生 SQL 语句
        - sql: SQL 语句
        - params: 参数化查询的字典, 示例: {"占位符": "值"}, 占位符使用:开头, 如:name, :age
        - is_return_data: 是否返回数据, 默认为 False, 如果为 True, 返回查询结果, 如果为 False, 返回空列表
        - return: Result 对象, 包含执行状态和结果
        #### 备注:
        - 支持参数化查询
        - 支持返回查询结果
        """
        try:
            # 检查 SQL 语句是否为空
            value_not_none(sql, "SQL 语句")
            # 如果 params 为 None, 则设置为空字典, 避免后续检查失败
            params = params if params is not None else {}
            # 检查参数是否为字典
            check_value_type(params, dict, "参数")
            # 检查数据库连接是否成功
            value_not_none(self.engine, is_diy=True, diy_msg="数据库连接失败")
            # 检查会话是否成功
            value_not_none(self.session, is_diy=True, diy_msg="数据库会话失败")
            # 检查数据库引擎是否成功
            value_not_none(self.engine.engine, is_diy=True, diy_msg="数据库引擎失败")
            # 检查数据库方言是否成功
            value_not_none(
                self.engine.engine.dialect, is_diy=True, diy_msg="数据库方言失败"
            )
            # 使用上下文管理器自动管理事务
            with self.session.begin():
                # 执行 SQL 语句
                result = self.session.execute(text(sql), params)
                # 获取查询结果
                rows = result.fetchall() if is_return_data else []
            # 返回 Result 对象
            return Result(status=True, result=rows, rowcount=result.rowcount)
        except Exception as e:
            self.session.rollback()
            raise SqlxDatabaseError(
                f"执行 SQL 语句失败: {e}\nSQL: {sql}\nParams: {params}"
            )
        finally:
            if self._enable_logger:
                self._logger.debug(f"执行 SQL 语句: {sql}")

    # close 方法, 用于关闭数据库连接和连接池
    def close(self):
        """
        #### 关闭数据库连接和连接池, 程序退出时调用
        """
        try:
            # 关闭会话
            self.session.close()
            # 关闭连接池
            self.engine.dispose()
        except Exception as e:
            # 如果关闭失败, 打印日志
            if self._enable_logger:
                self._logger.error(f"关闭数据库连接失败: {e}")
            # 抛出异常, 并包装异常信息
            raise SqlxDatabaseError(f"关闭数据库连接失败: {e}")

    # 静态方法 newSqlx, 用于创建 Sqlx 实例
    @staticmethod
    def newSqlx(
        db_type: str,
        host: str,
        port: int,
        user: str,
        password: str,
        db_name: str,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_timeout: int = 30,
        pool_recycle: int = 3600,
        enable_log: bool = False,
    ):
        """
        #### 创建并返回 Sqlx 实例
        - db_type: 数据库类型(如 'mysql', 'postgresql', 'sqlite') sqlite 支持自动创建数据目录
        - host: 数据库主机地址
        - port: 数据库端口
        - user: 数据库用户名
        - password: 数据库密码
        - db_name: 数据库名称
        - pool_size: 连接池大小
        - max_overflow: 最大溢出连接数
        - pool_timeout: 获取连接的超时时间
        - pool_recycle: 连接的最大生命周期
        - enable_log: 是否启用日志
        - return: Sqlx 实例
        """
        try:
            # 检查数据库类型是否为空
            value_not_none(db_type, "数据库类型")

            # 检查数据库连接参数是否为空
            if not host or not port or not user or not password:
                # 如果数据库类型是sqlite, 则不需要连接参数
                if not db_type == "sqlite":
                    raise ValueError("数据库连接参数不能为空")

            # 检查数据库名称是否为空
            value_not_none(
                db_name,
                is_diy=True,
                diy_msg="数据库名称不能为空, 如果是sqlite, 则为数据库文件路径",
            )

            # 构建数据库连接字符串, 如果是sqlite, 则直接使用文件路径
            if db_type == "sqlite":
                db_url = f"sqlite:///{db_name}"
                # 检查数据库文件父目录是否存在
                if not os.path.exists(os.path.dirname(db_name)):
                    os.makedirs(os.path.dirname(db_name))

            # 如果是mysql, 则使用mysql+pymysql连接
            elif db_type == "mysql":
                db_url = f"{db_type}://{user}:{password}@{host}:{port}/{db_name}"

            # 如果是postgresql, 则使用postgresql+psycopg2连接
            elif db_type == "postgresql":
                db_url = f"{db_type}://{user}:{password}@{host}:{port}/{db_name}"

            else:
                raise ValueError(f"不支持的数据库类型: {db_type}")

            # 检查数据库连接字符串是否为空
            value_not_none(db_url, "数据库连接字符串")

            # 检查连接池大小是否为空
            value_not_none(pool_size, "连接池大小")

            # 检查最大溢出连接数是否为空
            value_not_none(max_overflow, "最大溢出连接数")

            # 检查获取连接的超时时间是否为空
            value_not_none(pool_timeout, "获取连接的超时时间")

            # 检查连接的最大生命周期是否为空
            value_not_none(pool_recycle, "连接的最大生命周期")

            # 检查是否启用日志标志是否为空
            value_not_none(enable_log, "是否启用日志标志")

        except Exception as e:
            # 抛出异常, 并包装异常信息
            raise SqlxDatabaseError(f"创建 Sqlx 实例失败: {e}")
        else:
            # 创建并返回 Sqlx 实例
            return Sqlx(
                db_url=db_url,
                pool_size=pool_size,
                max_overflow=max_overflow,
                pool_timeout=pool_timeout,
                pool_recycle=pool_recycle,
                enable_log=enable_log,
            )

    def exec_script(self, script_path: str, batch_size: int = 100) -> Result:
        """
        #### 执行 SQL 脚本文件
        - script_path: SQL 脚本文件路径
        - batch_size: 每次执行的 SQL 语句数量, 默认为 100, 不能小于等于0
        - return: 返回 Result 对象, 包含执行状态和影响行数
        #### 注意点:
        - 自动事务管理
        - 支持分批执行 SQL 语句
        """
        # 总影响行数
        total_rows_affected = 0

        try:
            # 检查 SQL 脚本文件路径是否为空
            value_not_none(script_path, "SQL 脚本文件路径")

            # 检查 SQL 脚本文件是否存在
            if not os.path.isfile(script_path):
                raise ValueError("SQL 脚本文件不存在")

            # 检查 batch_size 是否为空
            value_not_none(batch_size, "每次执行的 SQL 语句数量")

            # 检查 batch_size 是否小于等于0
            if batch_size <= 0:
                raise ValueError("每次执行的 SQL 语句数量必须大于0")

            # 先尝试用 gbk 编码读取文件, 报错的话再用 utf-8 编码读取文件
            try:
                with open(script_path, "r", encoding="gbk") as file:
                    sql_script = file.read()
            except UnicodeDecodeError:
                try:
                    with open(script_path, "r", encoding="utf-8") as file:
                        sql_script = file.read()
                except Exception as e:
                    raise ValueError(f"SQL 脚本文件读取失败: {e}")
            except Exception as e:
                raise ValueError(f"SQL 脚本文件读取失败: {e}")

            # 分隔 SQL 语句
            try:
                statements = sqlparse.split(sql_script)  # 使用 sqlparse 解析 SQL 语句
            except Exception as e:
                raise ValueError(f"SQL 脚本文件解析失败: {e}")

            # 移除空语句
            statements = [stmt.strip() for stmt in statements if stmt.strip()]

            # 计算总批次
            total_batches = (len(statements) + batch_size - 1) // batch_size

            # 从 start_batch 开始执行F
            for batch_idx in range(total_batches):
                start = batch_idx * batch_size  # 计算当前批次的起始索引
                end = start + batch_size  # 计算当前批次的结束索引
                current_batch = statements[start:end]  # 获取当前批次的 SQL 语句

                # 使用上下文管理器自动管理事务
                with self.session.begin():
                    # 执行当前批次的 SQL 语句
                    count = 0
                    # 执行当前批次的 SQL 语句
                    for statement in current_batch:
                        result = self.session.execute(text(statement))
                        if self._enable_logger:
                            self._logger.debug(f"执行 SQL 语句: {statement}")
                        count += result.rowcount  # 累加影响行数
                    total_rows_affected += count  # 累加到总行数

                # 记录成功执行的批次
                if self._enable_logger:
                    self._logger.success(
                        f"批次 {batch_idx + 1}/{total_batches} 执行完成，影响行数: {count}"
                    )

            # 返回 Result 对象
            return Result(status=True, result=None, rowcount=total_rows_affected)

        # 捕获异常
        except Exception as e:
            self.session.rollback()  # 回滚事务
            if self._enable_logger:
                self._logger.error(f"执行 SQL 脚本文件失败: {e}")
            raise SqlxDatabaseError(f"执行 SQL 脚本文件失败: {e}")

    def get_connection_info(self) -> dict:
        """
        #### 获取当前数据库连接信息
        - return: 返回一个字典, 包含数据库连接信息
        ##### 键说明：
        - status: 连接状态, True表示连接成功, False表示连接失败
        - db_url: 数据库连接字符串 示例: sqlite:///./data/app.db (相对路径)
        - pool_size: 连接池大小
        - max_overflow: 最大溢出连接数
        - pool_timeout: 获取连接的超时时间
        - pool_recycle:  连接的最大生命周期
        """
        if self.engine:
            return {
                "status": True,
                "db_url": self.engine.url,
                "pool_size": self.engine.pool.size(),
                "max_overflow": self.engine.pool.overflow(),
                "pool_timeout": self.engine.pool._timeout,
                "pool_recycle": self.engine.pool._recycle,
            }
        else:
            return {
                "status": False,
                "db_url": None,
                "pool_size": None,
                "max_overflow": None,
                "pool_timeout": None,
                "pool_recycle": None,
            }

    def truncate_table(self, table_name: str) -> bool:
        """
        #### 清空指定表数据
        - table_name: 表名
        - return: bool 类型, True 表示清空成功, False 表示清空失败
        """
        try:
            # 构建 TRUNCATE TABLE 语句
            truncate_sql = f"TRUNCATE TABLE {table_name};"
            # 执行 SQL 语句
            Result = self.exec_sql(truncate_sql)

            # 返回 Result 对象中的状态
            return Result.status

        except Exception as e:
            raise SqlxDatabaseError(f"清空表数据失败: {e}")
        finally:
            if self._enable_logger:
                self._logger.debug(f"执行 SQL 语句: {truncate_sql}")
                self._logger.debug(f"清空表数据: {table_name}")


class SqlTx:
    """
    #### SqlTx类, 用于封装数据库操作, 手动管理事务。
    #### 注意点:
    - 手动管理事务, 必须手动调用 commit() 或 rollback() 方法, 否则事务不会提交或回滚。
    - 手动管理事务, 必须手动调用 close() 方法, 否则连接池不会关闭。
    #### 方法
    - newSqlTx(db_url: str, pool_size: int = 5, max_overflow: int = 10, pool_timeout: int = 30, pool_recycle: int = 3600, enable_log: bool = False) -> SqlTx: 创建一个新的SqlTx对象
    - exec_sql(sql: str, params: List[Any] = None) -> Result: 执行 SQL 语句
    - exec_script(self, script_path: str, batch_size: int = 100) -> Result: 执行脚本文件
    - get_connection_info(self) -> dict: 获取数据库连接信息
    - begin(self) -> None: 开始事务
    - commit(self) -> None: 提交事务
    - rollback(self) -> None: 回滚事务
    - close(self) -> None: 关闭数据库连接
    """

    def __init__(
        self,
        db_url: str,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_timeout: int = 30,
        pool_recycle: int = 3600,
        enable_log: bool = False,
    ):
        """
        #### 初始化数据库连接和连接池
        - db_url: 数据库连接字符串
        - pool_size: 连接池中维护的最大连接数 默认值为 5
        - max_overflow: 连接池中允许超过 pool_size 的临时连接数 默认值为 10
        - pool_timeout: 获取数据库连接时等待的最长时间（秒）默认值为 30 秒
        - pool_recycle: 连接的最大生命周期（秒）, 超过这个时间的连接将被关闭并重新创建, 设置为 -1 表示不关闭连接, 默认值为 3600 秒
        - enable_log: 是否启用日志, 默认值为 False
        - return: SqlTx 对象
        #### 示例:
        1. sqlite3 数据库连接字符串: sqlite:///./data/app.db (相对路径)
        2. sqlite3 数据库连接字符串: sqlite:////data/app.db (绝对路径)
        3. mysql 数据库连接字符串: mysql://user:password@host:port/dbname
        4. postgresql 数据库连接字符串: postgresql://user:password@host:port/dbname
        """
        try:
            # 检查数据库连接字符串是否为空
            value_not_none(db_url, "数据库连接字符串")
            # 检查连接池大小是否为空
            value_not_none(pool_size, "连接池大小")
            # 检查最大溢出连接数是否为空
            value_not_none(max_overflow, "最大溢出连接数")
            # 检查连接超时时间是否为空
            value_not_none(pool_timeout, "连接超时时间")
            # 检查连接最大生命周期是否为空
            value_not_none(pool_recycle, "连接最大生命周期")
            # 检查连接池大小是否为正整数
            if pool_size <= 0:
                raise ValueError("连接池大小必须为正整数, 切不能小于等于0")
            # 检查最大溢出连接数是否为正整数
            if max_overflow <= 0:
                raise ValueError("最大溢出连接数必须为正整数, 切不能小于等于0")
            # 检查连接超时时间是否为正整数
            if pool_timeout <= 0:
                raise ValueError("连接超时时间必须为正整数, 切不能小于等于0")
            # 创建数据库引擎
            self.engine = create_engine(
                db_url,
                pool_size=pool_size,
                max_overflow=max_overflow,
                pool_timeout=pool_timeout,
                pool_recycle=pool_recycle,
            )
            # 创建会话工厂
            self.Session = sessionmaker(bind=self.engine)
            # 创建会话
            self.session = self.Session()
            # 设置是否启用日志
            self._enable_logger = enable_log
            # 事务状态跟踪标志 , 默认为 False
            self._transaction_active = False
            # 实例化日志器
            self._logger = Logger()

        except Exception as e:
            # 如果连接失败,将engine设置为None, engine是数据库引擎
            self.engine = None
            # 如果连接失败, 将Session设置为None, Session是会话工厂
            self.Session = None
            # 如果连接失败, 将session设置为None, session是会话
            self.session = None
            # 实例化失败, 抛出异常
            if self._enable_logger:
                self._logger.error(f"实例化Sqlx失败: {e}")
            raise (f"实例化Sqlx失败: {e}")

    # exec_sql 方法, 用于执行 SQL 语句
    def exec_sql(
        self, sql: str, params: dict = None, is_return_data: bool = False
    ) -> Result:
        """
        #### 执行原生 SQL 语句
        - sql: SQL 语句
        - params: 参数化查询的字典, 示例: {"占位符": "值"}, 占位符使用:开头, 如:name, :age
        - is_return_data: 是否返回数据, 默认为 False, 如果为 True, 返回查询结果, 如果为 False, 返回空列表
        - return: Result 对象, 包含执行状态和结果
        #### 备注:
        - 支持参数化查询
        - 支持返回查询结果
        """
        try:
            # 检查 SQL 语句是否为空
            value_not_none(sql, "SQL 语句")
            # 如果 params 为 None, 则设置为空字典, 避免后续检查失败
            params = params if params is not None else {}
            # 检查参数是否为字典
            check_value_type(params, dict, "参数")
            # 检查数据库连接是否成功
            value_not_none(self.engine, is_diy=True, diy_msg="数据库连接失败")
            # 检查会话是否成功
            value_not_none(self.session, is_diy=True, diy_msg="数据库会话失败")
            # 检查数据库引擎是否成功
            value_not_none(self.engine.engine, is_diy=True, diy_msg="数据库引擎失败")
            # 检查数据库方言是否成功
            value_not_none(
                self.engine.engine.dialect, is_diy=True, diy_msg="数据库方言失败"
            )

            # 手动管理事务
            # 执行 SQL 语句
            result = self.session.execute(text(sql), params)

            # 获取查询结果
            rows = result.fetchall() if is_return_data else []

            # 返回 Result 对象
            return Result(status=True, result=rows, rowcount=result.rowcount)

        except Exception as e:
            self.rollback()  # 回滚事务
            raise SqlxDatabaseError(
                f"执行 SQL 语句失败: {e}\nSQL: {sql}\nParams: {params}"
            )
        finally:
            if self._enable_logger:
                self._logger.debug(f"执行 SQL 语句: {sql}")

    # close 方法, 用于关闭数据库连接和连接池
    def close(self):
        """
        #### 关闭数据库连接和连接池, 程序退出时调用
        """
        try:
            # 关闭会话
            self.session.close()
            # 关闭连接池
            self.engine.dispose()
        except Exception as e:
            # 抛出异常, 并包装异常信息
            if self._enable_logger:
                self._logger.error(f"关闭数据库连接失败: {e}")
            raise SqlxDatabaseError(f"关闭数据库连接失败: {e}")

    # 静态方法 newSqlx, 用于创建 Sqlx 实例
    @staticmethod
    def newSqlTx(
        db_type: str,
        host: str,
        port: int,
        user: str,
        password: str,
        db_name: str,
        pool_size: int = 5,
        max_overflow: int = 10,
        pool_timeout: int = 30,
        pool_recycle: int = 3600,
        enable_log: bool = False,
    ):
        """
        #### 创建并返回 SqlTx 实例
        - db_type: 数据库类型(如 'mysql', 'postgresql', 'sqlite') sqlite 支持自动创建数据目录
        - host: 数据库主机地址
        - port: 数据库端口
        - user: 数据库用户名
        - password: 数据库密码
        - db_name: 数据库名称
        - pool_size: 连接池大小
        - max_overflow: 最大溢出连接数
        - pool_timeout: 获取连接的超时时间
        - pool_recycle: 连接的最大生命周期
        - enable_log: 是否启用日志
        - return: SqlTx 实例
        """
        try:
            # 检查数据库类型是否为空
            value_not_none(db_type, "数据库类型")

            # 检查数据库连接参数是否为空
            if not host or not port or not user or not password:
                # 如果数据库类型是sqlite, 则不需要连接参数
                if not db_type == "sqlite":
                    raise ValueError("数据库连接参数不能为空")

            # 检查数据库名称是否为空
            value_not_none(
                db_name,
                is_diy=True,
                diy_msg="数据库名称不能为空, 如果是sqlite, 则为数据库文件路径",
            )

            # 构建数据库连接字符串, 如果是sqlite, 则直接使用文件路径
            if db_type == "sqlite":
                db_url = f"sqlite:///{db_name}"
                # 检查数据库文件父目录是否存在
                if not os.path.exists(os.path.dirname(db_name)):
                    os.makedirs(os.path.dirname(db_name))

            # 如果是mysql, 则使用mysql+pymysql连接
            elif db_type == "mysql":
                db_url = f"{db_type}://{user}:{password}@{host}:{port}/{db_name}"

            # 如果是postgresql, 则使用postgresql+psycopg2连接
            elif db_type == "postgresql":
                db_url = f"{db_type}://{user}:{password}@{host}:{port}/{db_name}"

            else:
                raise ValueError(f"不支持的数据库类型: {db_type}")

            # 检查数据库连接字符串是否为空
            value_not_none(db_url, "数据库连接字符串")

            # 检查连接池大小是否为空
            value_not_none(pool_size, "连接池大小")

            # 检查最大溢出连接数是否为空
            value_not_none(max_overflow, "最大溢出连接数")

            # 检查获取连接的超时时间是否为空
            value_not_none(pool_timeout, "获取连接的超时时间")

            # 检查连接的最大生命周期是否为空
            value_not_none(pool_recycle, "连接的最大生命周期")

            # 检查是否启用日志标志是否为空
            value_not_none(enable_log, "是否启用日志标志")

        except Exception as e:
            # 抛出异常, 并包装异常信息
            raise SqlxDatabaseError(f"创建 Sqlx 实例失败: {e}")
        else:
            # 创建并返回 SqlTx 实例
            return SqlTx(
                db_url=db_url,
                pool_size=pool_size,
                max_overflow=max_overflow,
                pool_timeout=pool_timeout,
                pool_recycle=pool_recycle,
                enable_log=enable_log,
            )

    def exec_script(self, script_path: str, batch_size: int = 100) -> Result:
        """
        #### 执行 SQL 脚本文件
        - script_path: SQL 脚本文件路径
        - batch_size: 每次执行的 SQL 语句数量, 默认为 100, 不能小于等于0
        - return: 返回 Result 对象, 包含执行状态和影响行数
        #### 注意点:
        -手动管理事务
        - 支持分批执行 SQL 语句
        """
        # 总影响行数
        total_rows_affected = 0

        try:
            # 检查 SQL 脚本文件路径是否为空
            value_not_none(script_path, "SQL 脚本文件路径")

            # 检查 SQL 脚本文件是否存在
            if not os.path.isfile(script_path):
                raise ValueError("SQL 脚本文件不存在")

            # 检查 batch_size 是否为空
            value_not_none(batch_size, "每次执行的 SQL 语句数量")

            # 检查 batch_size 是否小于等于0
            if batch_size <= 0:
                raise ValueError("每次执行的 SQL 语句数量必须大于0")

            # 读取 SQL 脚本文件内容
            with open(script_path, "r") as file:
                sql_script = file.read()

            # 分隔 SQL 语句
            try:
                statements = sqlparse.split(sql_script)  # 使用 sqlparse 解析 SQL 语句
            except Exception as e:
                raise ValueError(f"SQL 脚本文件解析失败: {e}")

            # 移除空语句
            statements = [stmt.strip() for stmt in statements if stmt.strip()]

            # 计算总批次
            total_batches = (len(statements) + batch_size - 1) // batch_size

            # 从 start_batch 开始执行F
            for batch_idx in range(total_batches):
                start = batch_idx * batch_size  # 计算当前批次的起始索引
                end = start + batch_size  # 计算当前批次的结束索引
                current_batch = statements[start:end]  # 获取当前批次的 SQL 语句

                # 执行当前批次的 SQL 语句
                count = 0
                # 执行当前批次的 SQL 语句
                for statement in current_batch:
                    result = self.session.execute(text(statement))
                    if self._enable_logger:
                        self._logger.debug(f"执行 SQL 语句: {statement}")
                    count += result.rowcount  # 累加影响行数
                total_rows_affected += count  # 累加到总行数

                # 记录成功执行的批次
                if self._enable_logger:
                    self._logger.success(
                        f"批次 {batch_idx + 1}/{total_batches} 执行完成，影响行数: {count}"
                    )

            # 返回 Result 对象
            return Result(status=True, result=None, rowcount=total_rows_affected)

        # 捕获异常
        except Exception as e:
            self.rollback()  # 回滚事务
            if self._enable_logger:
                self._logger.error(f"执行 SQL 脚本文件失败: {e}")
            raise SqlxDatabaseError(f"执行 SQL 脚本文件失败: {e}")

    def get_connection_info(self) -> dict:
        """
        #### 获取当前数据库连接信息
        - return: 返回一个字典, 包含数据库连接信息
        ##### 键说明：
        - status: 连接状态, True表示连接成功, False表示连接失败
        - db_url: 数据库连接字符串 示例: sqlite:///./data/app.db (相对路径)
        - pool_size: 连接池大小
        - max_overflow: 最大溢出连接数
        - pool_timeout: 获取连接的超时时间
        - pool_recycle:  连接的最大生命周期
        """
        if self.engine:
            return {
                "status": True,
                "db_url": self.engine.url,
                "pool_size": self.engine.pool.size(),
                "max_overflow": self.engine.pool.overflow(),
                "pool_timeout": self.engine.pool._timeout,
                "pool_recycle": self.engine.pool._recycle,
            }
        else:
            return {
                "status": False,
                "db_url": None,
                "pool_size": None,
                "max_overflow": None,
                "pool_timeout": None,
                "pool_recycle": None,
            }

    def begin(self):
        """
        #### 开始一个事务
        """
        # 检查是否已经有事务开启
        if self._transaction_active:
            if self._enable_logger:
                self._logger.warning("事务已经开启，不能再次开启新的事务")
            raise SqlxDatabaseError("事务已经开启，不能再次开启新的事务")

        # 执行开始事务
        try:
            self.session.begin()
            self._transaction_active = True  # 设置事务状态为 True
            if self._enable_logger:
                self._logger.debug("事务开始")
        except Exception as e:
            if self._enable_logger:
                self._logger.error(f"开始事务失败: {e}")
            raise SqlxDatabaseError(f"开始事务失败: {e}")

    def commit(self):
        """
        #### 提交当前事务
        #### 注意事项：
        - 如果提交事务失败，会尝试回滚事务
        - 需要在事务结束后调用该方法
        - 需要在事务开始前调用 begin 方法
        """
        # 检查是否有事务开启
        if not self._transaction_active:
            if self._enable_logger:
                self._logger.warning("没有开启事务，无法提交")
            raise SqlxDatabaseError("没有开启事务，无法提交")

        # 执行提交
        try:
            self.session.commit()
            self._transaction_active = False  # 提交后重置事务状态
            if self._enable_logger:
                self._logger.success("事务提交成功")
        except Exception as e:
            try:
                self.rollback()  # 提交失败后回滚
            except Exception as rollback_e:
                # 如果 rollback 也失败，记录更详细的信息
                if self._enable_logger:
                    self._logger.error(f"提交事务失败后回滚失败: {rollback_e}")
                raise SqlxDatabaseError(f"提交事务失败后回滚失败: {rollback_e}")
            if self._enable_logger:
                self._logger.error(f"提交事务失败: {e}")
            raise SqlxDatabaseError(f"提交事务失败: {e}")

    def rollback(self):
        """
        #### 回滚当前事务
        """
        # 检查是否有事务开启
        if not self._transaction_active:
            if self._enable_logger:
                self._logger.warning("没有开启事务，无法回滚")
            raise SqlxDatabaseError("没有开启事务，无法回滚")

        # 执行回滚
        try:
            self.session.rollback()
            self._transaction_active = False  # 提交后重置事务状态
            if self._enable_logger:
                self._logger.success("事务回滚成功")
        except Exception as e:
            if self._enable_logger:
                self._logger.error(f"回滚事务失败: {e}")
            raise SqlxDatabaseError(f"回滚事务失败: {e}")


class Logger:
    """
    ### 日志记录器类，用于记录不同级别的日志信息。
    #### 属性:
    - log_level (str): 日志级别，默认为 "DEBUG"。
    - log_color (bool): 是否启用日志颜色，默认为 True。
    - log_bold (bool): 是否启用日志加粗，默认为 True。
    """

    LEVEL_MAPPING = {
        "DEBUG": 10,
        "INFO": 20,
        "SUCCESS": 30,
        "WARNING": 40,
        "ERROR": 50,
    }

    def __init__(
        self, log_level: str = "DEBUG", log_color: bool = True, log_bold: bool = True
    ):
        """
        ### 初始化 Logger 类的构造函数。
        #### 参数:
        - log_level (str): 日志级别，默认为 "DEBUG"。
        - log_color (bool): 是否启用日志颜色，默认为 True。
        - log_bold (bool): 是否启用日志加粗，默认为 True。
        """
        log_level = log_level.upper()  # 转换为大写
        if log_level not in self.LEVEL_MAPPING:
            raise ValueError(
                f"不支持的日志级别: {log_level}, 期望的级别: {list(self.LEVEL_MAPPING.keys())}"
            )
        self.log_level = log_level  # 日志级别
        self.log_level_num = self.LEVEL_MAPPING[log_level]
        self.log_color = log_color  # 是否启用日志颜色
        self.log_bold = log_bold  # 是否启用日志加粗
        self.log_color_mapping = {
            "DEBUG": 35,
            "INFO": 34,
            "SUCCESS": 32,
            "WARNING": 33,
            "ERROR": 31,
        }

    def _log(self, level, message, color, bold):
        """
        ### 内部方法，用于格式化和输出日志信息。
        #### 参数:
        - level (str): 日志级别，如 "DEBUG", "INFO", "SUCCESS", "WARNING", "ERROR"。
        - message (str): 要记录的日志信息。
        - color (int): 日志颜色的ANSI转义码。
        - bold (bool): 是否加粗日志文本。
        """
        try:
            # 获取当前时间
            now = datetime.datetime.now()
            time_str = now.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]

            # 获取调用此方法的函数信息
            try:
                caller_frame = inspect.stack()[2]
                file_path = caller_frame.filename
                function = caller_frame.function
                line_number = caller_frame.lineno
            except IndexError:
                file_path = None
                function = None
                line_number = None

            # 获取文件名 (去掉父路径和文件后缀)
            try:
                if file_path:
                    file_name = os.path.basename(file_path)
                    module_name = os.path.splitext(file_name)[0]
                else:
                    module_name = None
            except:
                module_name = None

            # 格式化日志来源部分
            source = f"{module_name or 'Unknown'}:{function or 'Unknown'}:{line_number or 'Unknown'}"

            # 格式化日志
            log_message = f"{time_str} | {level} | {source} - {message}"

            # 应用颜色和加粗
            color = self.log_color_mapping.get(level, 37)  # 默认颜色为灰色
            escape_code = f"\033[{color}"
            if self.log_bold and bold:
                escape_code += ";1m"
            else:
                escape_code += "m"

            # 根据是否启用颜色输出日志
            if self.log_color:
                print(f"{escape_code}{log_message}\033[0m")
            else:
                print(log_message)

        except Exception as e:
            print(f"日志记录失败: {e}")

    def debug(self, message):  # 定义一个名为debug的方法，用于记录调试信息
        level = "DEBUG"  # 设置日志级别为"DEBUG"
        if (
            self.LEVEL_MAPPING[level] >= self.log_level_num
        ):  # 检查当前日志级别是否允许记录DEBUG级别的日志
            self._log(
                level, message, 35, True
            )  # 如果允许，则调用内部日志记录方法_log，记录DEBUG级别的日志信息

    def info(self, message):
        level = "INFO"
        if self.LEVEL_MAPPING[level] >= self.log_level_num:
            self._log(level, message, 34, True)

    def success(self, message):
        level = "SUCCESS"
        if self.LEVEL_MAPPING[level] >= self.log_level_num:
            self._log(level, message, 32, True)

    def warning(self, message):
        level = "WARNING"
        if self.LEVEL_MAPPING[level] >= self.log_level_num:
            self._log(level, message, 33, True)

    def error(self, message):
        level = "ERROR"
        if self.LEVEL_MAPPING[level] >= self.log_level_num:
            self._log(level, message, 31, True)


def is_valid_sql(sql: str) -> bool:
    """
    #### 检查 SQL 语句的格式和语法是否正确
    - sql: SQL 语句
    - return: True 如果格式正确，否则 False
    """
    try:
        # 使用 sqlparse 解析 SQL 语句
        parsed = sqlparse.parse(sql)
        # 如果解析结果为空，说明 SQL 语句为空或无效
        if not parsed:
            return False
        # 如果解析成功且没有抛出异常，说明格式正确
        return True
    except Exception:
        # 如果解析过程中抛出异常，说明格式错误
        return False


def is_safe_sql(sql: str) -> bool:
    """
    #### 检查 SQL 语句是否存在危险操作
    - sql: SQL 语句
    - return: True 如果安全，否则 False
    """
    # 检查 SQL 语句是否为空
    if not sql:
        return False

    # 首先检查 SQL 语句的格式是否正确
    if not is_valid_sql(sql):
        return False

    # 定义危险关键字
    dangerous_keywords = [
        "UNION",
        "DELETE",
        "TRUNCATE",
        "DROP",
        "ALTER",
        "GRANT",
        "REVOKE",
    ]

    # 解析 SQL 语句
    parsed = sqlparse.parse(sql)[0]

    # 遍历解析后的 tokens
    for token in parsed.tokens:
        # 检查是否包含危险关键字
        if token.ttype is sqlparse.tokens.DML or token.ttype is sqlparse.tokens.DDL:
            if token.value.upper() in dangerous_keywords:
                return False

        # 检查是否包含危险的子句
        if isinstance(token, sqlparse.sql.Where):
            where_clause = sqlparse.format(
                str(token), strip_comments=True, reindent=True
            )
            for keyword in dangerous_keywords:
                if re.search(r"\b" + keyword + r"\b", where_clause, re.IGNORECASE):
                    return False

    # 如果没有发现危险关键字或子句，返回 True
    return True
